package com.lhb.zookeeper.service;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

public class LockSample {

    private ZooKeeper zkClient;
    private static final String LOCK_ROOT_PATH = "/Locks";
    private static final String LOCK_NODE_NAME = "Lock_";
    private String lockPath;

    public LockSample() throws IOException {
        zkClient = new ZooKeeper("localhost:2181", 10000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if (watchedEvent.getState() == Event.KeeperState.Disconnected){
                    System.out.println("失去连接");
                }
            }
        });
    }

    public void acquireLock() throws KeeperException, InterruptedException {
        //创建锁
        createLock();
        //尝试获取锁
        attemptLock();
    }

    private void attemptLock() throws KeeperException, InterruptedException {

        //获取所有的子节点
        List<String> lockPaths = null;
        lockPaths = zkClient.getChildren(LOCK_ROOT_PATH,false);
        Collections.sort(lockPaths);
        int index = lockPaths.indexOf(lockPath.substring(LOCK_ROOT_PATH.length()+1));

        //如果是最小节点
        if(index == 0){
            System.out.println(Thread.currentThread().getName()+"获得锁：lockPath"+lockPath);
            return;
        }else {
            //监听前一个节点
            String preLockPath = lockPaths.get(index - 1);
            Stat stat = zkClient.exists(LOCK_ROOT_PATH+"/"+preLockPath,watcher);
            //当前锁不存在，如执行完毕，当前执行节点掉线，重新获取锁
            if(stat == null){
                attemptLock();
            } else { //阻塞当前进程，直到preLockPath释放锁，被watcher监听到，notifyAll后，重新获取锁
                System.out.println(Thread.currentThread().getName()+"获得锁：lockPath"+lockPath);
                synchronized (watcher){
                    watcher.wait();
                }
                attemptLock();
            }
        }
    }

    private Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println(event.getPath() + "前锁释放");
            synchronized (this){
                notifyAll();
            }
        }
    };

    private void createLock() throws KeeperException, InterruptedException {
        Stat stat = zkClient.exists(LOCK_ROOT_PATH, false);
        if(stat == null){
            //创建根节点
            zkClient.create(LOCK_ROOT_PATH,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }
        //创建临时节点
        String lockPath = zkClient.create(LOCK_ROOT_PATH+"/"+LOCK_NODE_NAME,Thread.currentThread().getName().getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(Thread.currentThread().getName()+"锁创建"+lockPath);
        this.lockPath = lockPath;
    }
}
